package com.dtkavin.spark.rdd

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.{Logging, SparkConf, SparkContext}

/**
  * Created by IntelliJ IDEA.
  * Programmer : John Zn
  * Date : 2016/4/14 0014
  * Time : 14:31
  * Discribtion : 将ip，location等信息写入mysql，便于以后使用此  ip规则库
  *   注意：！！！本地是GBK编码，连接时需要更改编码为utf8--》jdbc:mysql://192.168.117.45/db_bigdata_rule?useUnicode=true&characterEncoding=UTF-8
  */
class Ip_location_jdbcRDD_RW extends Serializable with Logging {
  def getConnection(): Connection = {
    Class.forName("com.mysql.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://192.168.117.45/db_bigdata_rule?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
  }

  def write2msyql(): Unit = {
    val conf = new SparkConf()
    val sc = new SparkContext("local[3]", "IP_LOCAL", conf)
    //    (ip_start: String, ip_stop: String, ip_l_start: String, ip_l_stop: String, continent: String, country: String, province: String, city: String, net_type: String, net_code: String, en_country: String, en_symbol: String, longitude: String, latitude: String)
    def data2mysql(iterator: Iterator[(String, String, String, String, String, String, String, String, String, String, String, String, String, String)]): Unit = {
      var conn: Connection = null
      var pstat: PreparedStatement = null
      val sql = "insert into t_ip_location (ip_start,ip_stop,ip_l_start,ip_l_stop,continent,country,province,city,net_type,net_code,en_country,en_symbol,longitude,latitude) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?)"

      try {
        conn = getConnection()
        //这么写是无序的
        iterator.foreach {
          x => {
            pstat = conn.prepareStatement(sql)
            pstat.setString(1, x._1)
            pstat.setString(2, x._2)
            pstat.setString(3, x._3)
            pstat.setString(4, x._4)
            pstat.setString(5, x._5)
            pstat.setString(6, x._6)
            pstat.setString(7, x._7)
            pstat.setString(8, x._8)
            pstat.setString(9, x._9)
            pstat.setString(10, x._10)
            pstat.setString(11, x._11)
            pstat.setString(12, x._12)
            pstat.setString(13, x._13)
            pstat.setString(14, x._14)
            pstat.executeUpdate()
          }
        }
      } catch {
        case e: Exception => e.printStackTrace()
      } finally {
        if (pstat != null) {
          pstat.close()
        }
        if (conn != null) {
          conn.close()
        }
      }
      logInfo("----> write done")
    }
    sc.textFile("D:\\CZ\\FQDownload\\feiq\\Recv Files\\ip.txt", 3).map {
      //此处的map和mapPartition有什么区别？？？
      x => {
        val line = x.split("\\|")
        (line(0), line(1), line(2), line(3), line(4), line(5), line(6), line(7), line(9), line(10), line(11), line(12), line(13), line(14))
      }
    }.foreachPartition(data2mysql(_))

    sc.stop()
  }

  /*
  * 从数据库获取ip信息
  * */
  //    def readFromMysql(): JdbcRDD ={
  //      new JdbcRDD[]()
  //    }
}

object Ip_location_jdbcRDD_RW {
  def main(args: Array[String]) {
    val jdbcRDD = new Ip_location_jdbcRDD_RW
    jdbcRDD.write2msyql()
  }
}